package david.java.flink_sql.udf.offline;

import lombok.Data;
import org.apache.hadoop.hive.ql.exec.UDF;

import java.io.Serializable;
import java.util.*;

/**
 * @Description: 本udf函数是用来计算表血缘关系的, 该方法遍历方式为 DFS, 不适合大数量, 递归层数较深的场景. 如果有大数据量或递归层数较深的场景,
 * 请不要使用本方法, 请使用BFS遍历方式实现
 * @Author: zhaodawei
 * @Date: Create in 3:37 下午 2022/3/8
 */
public class GetBloodLength extends UDF {

    public List<String> evaluate(List<String> list) {
        Map<String, HashSet<String>> forword = new HashMap<String, HashSet<String>>();
        Map<String, HashSet<String>> back = new HashMap<String, HashSet<String>>();
        Set<String> words = new HashSet<>();
        List<String> result = new ArrayList<>();

        for (String value : list) {
            String[] strs = value.split("-");
            if (forword.containsKey(strs[0])) {
                HashSet<String> set = forword.get(strs[0]);
                set.add(strs[1]);
                forword.put(strs[0], set);
            } else {
                HashSet<String> set = new HashSet<>();
                set.add(strs[1]);
                forword.put(strs[0], set);
            }

            if (back.containsKey(strs[1])) {
                HashSet<String> set = back.get(strs[1]);
                set.add(strs[0]);
                back.put(strs[1], set);
            } else {
                HashSet<String> set = new HashSet<>();
                set.add(strs[0]);
                back.put(strs[1], set);
            }

            words.add(strs[0]);
            words.add(strs[1]);
        }

        for (String word : words) {
            BloodLength bloodLength = new BloodLength();
            bloodLength.setName(word);
            bloodLength.setEnd(recursive(forword, word, new HashSet<String>()));
            bloodLength.setTop(recursive(back, word, new HashSet<String>()));
            bloodLength.build();
            // 临时改成map: 为了减少长度, 自定义name(还有就是用fastjson转成的string 总是多一个bloodLength,什么鬼~)
            Map<String, Object> map = new HashMap<>();
            map.put("name", bloodLength.getName());
            map.put("top", bloodLength.getTop());
            map.put("end", bloodLength.getEnd());
            map.put("length", bloodLength.getLength());
            result.add(toJsonString(map));
        }

        return result;
    }

    public Integer recursive(Map<String, HashSet<String>> data, String value, HashSet<String> storedKeys) {
        HashSet<String> valueSet = data.get(value);
        storedKeys.add(value);
        if (valueSet == null) {
            return 0;
        }
        int maxDeep = 0;
        for (String str : valueSet) {
            // 排除掉环
            if (!storedKeys.contains(str)) {
                maxDeep = Math.max(recursive(data, str, storedKeys) + 1, maxDeep);
            }
        }

        return maxDeep;
    }


    public static void main(String[] args) {
        GetBloodLength gts = new GetBloodLength();
        List<String> list1 = new ArrayList<String>();
        list1.add("a-b");
        list1.add("b-a");
        list1.add("b-c");
        list1.add("c-b");
        list1.add("c-d");
        list1.add("d-c");
        System.out.println(gts.evaluate(list1));
        // System.out.println();

        List<String> list2 = new ArrayList<String>();
        list2.add("a-b");
        list2.add("b-c");
        list2.add("c-a");
        System.out.println(gts.evaluate(list2));
        System.out.println();


        List<String> list3 = new ArrayList<String>();
        list3.add("a-b");
        list3.add("b-c");
        list3.add("c-d");
        list3.add("e-c");
        list3.add("f-e");
        list3.add("g-f");
        System.out.println(gts.evaluate(list3));
    }


    public static String toJsonString(Map<String, Object> map) {
        StringBuilder sb = new StringBuilder();
        sb.append("{");
        for (Map.Entry<String, Object> m: map.entrySet()) {
            String name = m.getKey();
            String value = String.valueOf(m.getValue());

            sb.append("\"").append(name).append("\"")
                    .append(":")
                    .append("\"").append(value).append("\"")
                    .append(",");
        }
        sb.deleteCharAt(sb.length()-1);
        sb.append("}");
        return sb.toString();
    }


    @Data
    public static class BloodLength implements Serializable {
        /**
         * 距离最远父顶点的距离
         */
        private int top;
        /**
         * 距离最远末端的距离
         */
        private int end;
        private int length;
        private String name;

        public int build() {
            this.length = top + end + 1;
            return this.length;
        }
    }
}





